package com.haima.sage.bigdata.api.common;


import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.haima.sage.bigdata.api.entity.storage.EsStorageData;
import com.haima.sage.bigdata.api.entity.storage.Host;
import com.haima.sage.bigdata.api.entity.storage.Storage;
import com.haima.sage.bigdata.api.exception.ValidateException;
import com.haima.sage.bigdata.api.vo.OrderParam;
import com.haima.sage.bigdata.api.vo.param.AggParam;
import com.haima.sage.bigdata.api.vo.param.agg.Agg;
import com.haima.sage.bigdata.api.vo.param.agg.AggType;
import com.haima.sage.bigdata.api.vo.param.group.Having;
import com.haima.sage.bigdata.api.vo.param.group.HavingOperatorType;
import com.haima.sage.bigdata.api.vo.param.order.OrderType;
import com.haima.sage.bigdata.api.vo.param.where.WhereCondition;
import com.haima.sage.bigdata.api.vo.param.where.WhereConditionGroup;
import com.haima.sage.bigdata.api.vo.param.where.WhereConditionItem;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;

import java.io.IOException;
import java.util.*;

public class EsUtils {
    private static final Logger LOG = LogManager.getLogger(EsUtils.class);
    private static final Map<String, RestClient> map = new HashMap<>();
    private Map<String, String> esMap = new HashMap<>();

    public static synchronized RestClient getClient(String store) {
        return map.get(store);
    }

    public static void merge(Storage storage) throws IOException {
        RestClient ds = map.get(storage.getId());
        if (ds != null) {
            // TODO 清理数据源
            Thread thread = new Thread(new CleanHandle(ds));
            thread.setDaemon(true);
            thread.start();
        }
        map.put(storage.getId(), create(storage));
    }

    private static RestClient create(Storage storage) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        EsStorageData map = mapper.readValue(storage.getData(), EsStorageData.class);
        if (map == null || map.getClusterHosts() == null || map.getClusterHosts().size() == 0) {
            throw new IllegalArgumentException("es storage config error");
        }
        List<HttpHost> hosts = new ArrayList<>();
        for (Host host : map.getClusterHosts()) {
            int port = 9200;
            if (host.getPort() != null) {
                port = host.getPort();
            }
            String schema = "http";
            if (!StringUtils.isBlank(host.getSchema())) {
                schema = host.getSchema();
            }
            HttpHost item = new HttpHost(host.getHost(), host.getPort(), schema);
            hosts.add(item);
        }
        return RestClient.builder(hosts.toArray(new HttpHost[hosts.size()])).build();
    }

    public void appendField(StringBuilder sb, List<String> fields){
        Iterator iterator = fields.iterator();
        String str = iterator.next().toString();
        if(fields != null && !str.equals("*")) {
            sb.append("\"_source\":[");
            Iterator<String> fieldz = fields.listIterator();
            while (fieldz.hasNext()) {
                String field = fieldz.next();
                sb.append("\"").append(field).append("\",");
            }
            sb.deleteCharAt(sb.length() - 1);
            sb.append("],");
        }else {
            // 所有字段
            //do nothing
        }
    }

    public void appendOffset(StringBuilder sb, Integer offset){
        if(offset != null ){
            sb.append("\"from\":").append(offset).append(",");
        }else{
            //do nothing
        }
    }

    public void appendLimit(StringBuilder sb, Integer limit, Integer maxBatchSize){
        if(limit != null && limit < maxBatchSize){
            sb.append("\"size\":").append(limit).append(",");
        }else if (limit != null && limit > maxBatchSize){
            sb.append("\"size\":").append(maxBatchSize).append(",");
        }else{
            // do nothing
        }
    }

    public void appendWhere(StringBuilder sb, String table, Storage storage, WhereCondition where) throws Exception{
        if (where instanceof WhereConditionGroup) {
            WhereConditionGroup group = (WhereConditionGroup) where;
            if (group.getConditions() != null && group.getConditions().size() > 0) {
                String Oper = group.getOperator().toString();
                if (("AND").equals(Oper)){
                    sb.append("{\"bool\":{").append("\"must\":[");
                }else if(Oper.equals("NOT")){
                    sb.append("{\"bool\":{").append("\"must_not\":[");
                }else {   // or
                    sb.append("{\"bool\":{").append("\"should\":[");
                }

                //遍历组内层数据
                for (WhereCondition item : group.getConditions()) {
                    if(item instanceof WhereConditionItem){
                    sb.append("{").append("\"bool\":{");
                    switch (item.getOperator()) {
                        case Eq:
                            if ("".equals(((WhereConditionItem) item).getValue())) {
                                sb.append("\"should\":[");
                            } else {
                                sb.append("\"must\":[");
                            }
                            break;
                        case Ne:
                                sb.append("\"must_not\":[");
                            break;
                        case Gt:
                            sb.append("\"must\":[");
                            break;
                        case Ge:
                            sb.append("\"must\":[");
                            break;
                        case Lt:
                            sb.append("\"must\":[");
                            break;
                        case Le:
                            sb.append("\"must\":[");
                            break;
                        case Like:
                            sb.append("\"must\": [");
                    }

                    appendWhere(sb, table, storage, item);
                    sb.append("]}},");
                }else{
                        appendWhere(sb, table, storage, item);
                    }
                }
                sb.deleteCharAt(sb.length() - 1);
                sb.append("]}},");
            }
        } else {
            WhereConditionItem item = (WhereConditionItem) where;
            appendWhereItem(table, item, sb, storage);
        }
    }


    public void appendWhereItem(String table, WhereConditionItem item , StringBuilder sb, Storage storage) throws Exception {
        Object value;
        value = transField(item.getValue());
        String field = getMappingTransField(item.getField());
        switch (item.getOperator()){
            case Eq:
                if("".equals(item.getValue().toString())){
                    sb.append("{\"bool\": {\"must_not\": [{\"exists\": {\"field\": \"").append(field).append("\"}}]}},");
                    sb.append("{\"bool\": {\"must\": [{\"term\": {\"").append(field).append("\": \"\"}} ]}}");
                } else {
                    sb.append("{\"term\":{\"")
                            .append(field).append("\":\"").append(value).append("\"")
                            .append("}}");
                }
                break;
            case Ne:
                sb.append("{\"term\":{\"")
                        .append(field).append("\":\"").append(value).append("\"")
                        .append("}}");
                if ("".equals(item.getValue())) {
                    sb.append("]}},{\"bool\": {\"must\": [{\"exists\": {\"field\": \"").append(field).append("\"}}");
                }
                break;
            case Gt:
                sb.append("{").append("\"range\":{\"")
                        .append(field).append("\":{")
                        .append("\"gt\":\"").append(value)
                        .append("\"}}}");
                break;
            case Ge:
                sb.append("{").append("\"range\":{\"")
                        .append(field).append("\":{")
                        .append("\"gte\":\"").append(value)
                        .append("\"}}}");
                break;
            case Lt:
                sb.append("{").append("\"range\":{\"")
                        .append(field).append("\":{")
                        .append("\"lt\":\"").append(value)
                        .append("\"}}}");
                break;
            case Le:
                sb.append("{").append("\"range\":{\"")
                        .append(field).append("\":{")
                        .append("\"lte\":\"").append(value)
                        .append("\"}}}");
                break;
            case Like:
                String likeValue = value.toString().replace("%","*"); //统一mysql % 与ES *
                sb.append("{").append("\"wildcard\": {\"")
                        .append(field).append("\": {")
                        .append("\"value\": \"").append(likeValue)
                        .append("\"}}}");
        }

    }

    public void appendOrderBy(StringBuilder sb, OrderParam param, Storage storage) throws Exception {
        Map<String, OrderType> orderBy = param.getOrderBy();
        String table = param.getTable();
        if(orderBy != null) {
            String field;
            sb.append("\"sort\":[{");
            for (Map.Entry<String, OrderType> orderType:orderBy.entrySet()) {
                String fieldp = orderType.getKey();   //要转换的字段
                field = getMappingTransField(fieldp);
                sb.append("\"").append(field).append("\":{")
                        .append("\"order\":\"").append(orderType.getValue()).append("\"")
                        .append("},");
            }
            sb.deleteCharAt(sb.length() - 1);
            sb.append("}],");
        }
    }

    public String getMappingTransField(String field) throws Exception {
            String value = esMap.get(field);
            if (value == null) throw new ValidateException(Constants.NOT_FOUND_DATA_ERROR_EN,
                  "[" + field + "] " + Constants.NOT_FOUND_DATA_ERROR_CN);
            return "keyword".equals(value) ? field + "." + value : field;

    }

    //List<String> groupBys, List<Agg> aggstat, Having having
    public void appendGroupby(StringBuilder sb, AggParam param, Storage storage) throws Exception {
        int limit = param.getLimit();
        /* 获取满足条数且尽可能少的数据
        if (param.getGroupBys()!=null){
            int gbSize = param.getGroupBys().size();
            int size = (int)
                    Math.ceil(StrictMath.pow((param.getOffset() + param.getLimit()), 1.0/ gbSize));
            limit = gbSize > 3 ? size + 1: size + 2;
        } */
        String table = param.getTable();
        List<String> groupBys = param.getGroupBys();
        if(groupBys != null) {
            for (int i = 1; i <= groupBys.size(); i++) {
                String toField = groupBys.get(i - 1);
                String field = getMappingTransField(toField);
                sb.append("\"aggs\": {")
                        .append("\"").append(groupBys.get(i - 1)).append("\"").append(": {")
                        .append("\"terms\": {")
                        .append("\"size\": ")
                        .append(limit)
                        .append(", \"field\": \"").append(field).append("\"},");
                if (i == groupBys.size()) {
                    appendAggOrder(sb, param, storage);
                    appendAgg(sb, param, storage);
                }
            }

            for (int i = 1; i <= groupBys.size(); i++) {
                sb.append("}}");

                if (i == groupBys.size()) {
                    sb.append(",");
                }
            }
        }else{
            appendAgg(sb, param, storage);
            sb.deleteCharAt(sb.length() - 1);
            sb.append("}}");
        }
    }

    public void appendAggOrder(StringBuilder sb, AggParam param, Storage storage) throws Exception {
        Map<String, OrderType> orders = param.getOrderBy();
        if (orders == null || orders.isEmpty()) return;
        sb.deleteCharAt(sb.length() - 2);
        sb.append("\"order\": [");
        for (Map.Entry<String, OrderType> entry : orders.entrySet()) {
            sb.append("{\"").append(entry.getKey()).append("\": \"").append(entry.getValue().getValue()).append("\"},");
        }
        sb.deleteCharAt(sb.length() - 1);
        sb.append("]},");
    }

    //List<Agg> aggstat, Having having
    public void appendAgg(StringBuilder sb, AggParam param, Storage storage) throws Exception {
        String table = param.getTable();
        boolean repeatAgg = false;
        List<Agg> aggstat = param.getAggs();
        String havingAggType;
        Having having;

        String aggName = null;
        sb.append("\"aggs\": {");
        Iterator iter = aggstat.iterator();
        while (iter.hasNext()) { //遍历es聚合操作
            Agg agg = (Agg) iter.next();
            String aggType = aggTypeTrans(agg.getType());
            String toField = agg.getField();
            String field = getMappingTransField(toField);
            if (param.getHaving() != null && agg.getField().equals(param.getHaving().getField())) {
                if (param.getHaving().getType() == null
                        ||(param.getHaving().getType() != null
                        && agg.getType().equals(param.getHaving().getType()))) {
                    repeatAgg = true;
                    aggName = agg.getName();
                }
            }
            sb.append("\"").append(agg.getName()).append("\": {")
                    .append("\"").append(aggType).append("\": {")
                    .append("\"field\": \"").append(field).append("\"")
                    .append("}},");
        }

        if (param.getHaving() != null) {
            having = param.getHaving();
            if (repeatAgg) {
                havingFilter(sb, aggName, having);
            } else if (param.getHaving().getType() != null) {
                havingAggType = aggTypeTrans(having.getType());
                sb.append("\"").append(havingAggType).append("\": {")
                        .append("\"").append(havingAggType).append("\": {")
                        .append("\"field\": \"").append(getMappingTransField(having.getField())).append("\"")
                        .append("}},");
                havingFilter(sb, having);
            }else {
                String name = having.getField();
                havingFilter(sb, name, having);

            }
        }
        sb.deleteCharAt(sb.length() - 1);
        sb.append("}");
    }
    // 添加having 过滤条件
    private void havingFilter(StringBuilder sb, String type, Having h){
        sb.append("\"bucket_filter\": {")
                .append("\"bucket_selector\": {")
                .append("\"buckets_path\": {")
                .append("\"value\": \"").append(type).append("\"")
                .append("},\"script\": \"params.value").append(havingTypeTrans(h.getOperator())).append(h.getValue()).append("\"}},");

    }

    // 添加having 过滤条件
    private void havingFilter(StringBuilder sb, Having h){
        String havingAggType = aggTypeTrans(h.getType());
        havingFilter(sb, havingAggType, h);
    }


    //把枚举字母计算类型转换为的运算符( > < = )
    public String havingTypeTrans(HavingOperatorType havingOperatorType){
        switch (havingOperatorType){   //Eq("="), Ne("!="), Gt(">"), Ge(">="), Lt("<"), Le("<=");
            case Eq:
                return "==";
            case Ne:
                return "!=";
            case Gt:
                return ">";
            case Ge:
                return ">=";
            case Lt:
                return "<";
            case Le:
                return "<=";
        }
        return null;
    }
    //把业务系统上传的聚合计算类型转换成ES能识别的聚合计算类型
    public String aggTypeTrans(AggType aggType){

        switch (aggType){ //Max("max"), Min("min"), Sum("sum"), Avg("avg"), Count("count"), CountDistinct("countdistinct");
            case Max:
                return "max";
            case Min:
                return "min";
            case Sum:
                return "sum";
            case Avg:
                return "avg";
            case Count:
                return "value_count";
            case CountDistinct:
                return "cardinality";
        }
        return null;
    }
    // 将查询字符串value中的单引号去掉
    public Object transField(Object obj){
        Object field;
        if(obj instanceof String){
            String objString = (String) obj;
            if (objString.length()>0 && objString.charAt(0) == '\'' && (objString.charAt(objString.length()-1) == '\'' )){
                field = objString.substring(1,objString.length()-1);
            }else {
                field = obj;
            }
        }else{
            field = obj;
        }
        return field;
    }

    //ES请求
    public List<Map> esPost(Storage storage, String table, Integer maxBatchSize, String entity, OrderParam param) throws Exception {
        String method = "POST";
        String uri = "/" + table + "/_search";
        Map<String, String> params = Collections.emptyMap();
        String key = storage.getId();

        RestClient client = EsUtils.getClient(key);
        if (client == null) {
            throw new Exception("can't get client");
        }
        Response response = null;
        if (StringUtils.isBlank(entity)) {
            response = client.performRequest(method, uri, params);
        } else {
            HttpEntity httpEntity = new NStringEntity(entity, ContentType.APPLICATION_JSON);
            response = client.performRequest(method, uri, params, httpEntity);
        }
        if(response.getStatusLine().getStatusCode() / 100 == 2) {
            List<String> fields = param.getFields();
            String responseBody = EntityUtils.toString(response.getEntity());
            ObjectMapper jackson = new ObjectMapper();
            //json字符串读成json树，get出来指定的节点
            JsonNode jsonNode = jackson.readTree(responseBody);
            JsonNode hits = jsonNode.get("hits").get("hits");
            Iterator<JsonNode> iterator = hits.elements();
            List<Map> list = new LinkedList<Map>();
            while (iterator.hasNext()) {
                Map<String, Object> map = jackson.readValue(iterator.next().get("_source").toString(), Map.class);
                if (fields == null || fields.get(0).equals("*")) {
                    fieldMakeUp(esMap.keySet().iterator(), map);
                }else {
                    fieldMakeUp(fields.iterator(), map);
                }
                list.add(map);
            }
            return list;
        }
        return null;
    }

    public boolean esMapping (String table, String pattern, Storage storage) throws Exception {
        Map<String, String> params = Collections.emptyMap();
        String key = storage.getId();
        RestClient client = EsUtils.getClient(key);
        // 支持 * 号查询
        String index = table;
        if(table.endsWith("*")){
            String prefix = table.substring(0, table.length()-1);
            DateUtils du = null;
            try {
                du = new DateUtils(pattern);
            } catch (Exception e){
                throw new Exception("table pattern '"+ pattern +"' must be date");
            }
            int code = 404;
            String date = du.today();
            for (int i=0; i< 10; i++) {
                index = prefix + date;
                Response response = client.performRequest("HEAD", index, params);
                if (response.getStatusLine().getStatusCode() == 200) {
                    code = 200;
                    break;
                }
                date = du.previous(date);
            }
            if (code != 200) throw new ValidateException("index "+ table +" mappings can not find!",
                    "索引"+ table +"未匹配到已存在的索引");
        }
        String uri = "/" + index + "/_mapping";
        if (client == null) throw new Exception("can't get client");
        Response response = client.performRequest("GET", uri, params);
        if(response.getStatusLine().getStatusCode() / 100 == 2) {
            String responseBody = EntityUtils.toString(response.getEntity(), "UTF-8");
            JsonNode jsonNode = new ObjectMapper().readTree(responseBody);
            JsonNode tableNameJsonNode = jsonNode.get(index);
            try {
                JsonNode mappings = tableNameJsonNode.get("mappings").get("data").get("properties");
                Iterator<String> fields = mappings.fieldNames();
                while (fields.hasNext()){
                    String field = fields.next();
                    JsonNode type = mappings.get(field).get("type");
                    if (type == null) {
                        LOG.error("Can not recognize Field '" + field + "' !");
                        continue;
                    }
                    String value = type.asText();
                    esMap.put(field, "text".equals(value) ? "keyword" : value);
                }
            } catch (Exception e) {
                throw new Exception("index '"+ index +"' mapping error, properties can not be found!");
            }
            return true;
        }
        return false;
    }

    private void fieldMakeUp(Iterator<String> fields, Map<String, Object> map){
        while (fields.hasNext()){
            String field = fields.next();
            if(!map.keySet().contains(field)){
                switch (esMap.getOrDefault(field, "keyword")) {
                    case "keyword":
                        map.put(field, "");
                        break;
                    case "Long":
                        map.put(field, 0);
                        break;
                    case "Integer":
                        map.put(field, 0);
                        break;
                    case "date":
                        map.put(field, "");
                        break;
                    case "double":
                        map.put(field, 0.0);
                        break;
                    case "float":
                        map.put(field, 0.0);
                        break;
                    case "boolean":
                        map.put(field, "");
                        break;
                    default:
                        map.put(field, "");
                        break;
                }
            }
        }
    }

}
